package com.nian.message.rocketmq.producer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.jupiter.api.Test;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.UUID;

/**
 * 用途：
 * 开发：zhangmj
 * 日期：2020/4/10 0010 15:57
 */
@Slf4j
public class TestProducer {

    @Test
    public void send1() {

        TransactionMQProducer producer = new TransactionMQProducer("nian-message-service-producer-group-t");
        producer.setNamesrvAddr("192.168.203.132:9876");
        producer.setTransactionListener(new TransactionListener(){
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                try{
                    String jsonString = new String((byte[]) message.getBody());
                    log.info("事务监听：");
                    log.info("{}", jsonString);
                    return LocalTransactionState.COMMIT_MESSAGE;
                }catch (Exception e){
                    log.info("错误：{}", e);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt message) {
                try{
                    final JSONObject jsonObject = JSON.parseObject(new String((byte[]) message.getBody()));
                    log.info("事务ID 检查：{}",jsonObject);
                    //事务id
                    String txId = message.getTransactionId();
                    String msgId = message.getMsgId();
                    log.info("消息ID：{}, 事务ID: {} ", msgId, txId);
                    // 假设成功
                    int c = 1;
                    log.info("事务ID: {} 结果: {}", txId, c);
                    if (c > 0) {
                        return LocalTransactionState.COMMIT_MESSAGE;
                    } else {
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                }catch (Exception e){
                    e.printStackTrace();
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }
        });
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        try {
            producer.start();
            String[] tags = new String[]{"TagA", "TagB"};

            for (int i = 0; i < 3; i++) {

                String body = i + "--RocketMQ 生产者测试【事务】消息：时间： " + sdf.format(new Date()) + ".";
                Message msg = new Message("producer_topic_shiwu_test_", "one", body.getBytes());
                msg.setTransactionId(UUID.randomUUID().toString());
                TransactionSendResult sendResult = producer.sendMessageInTransaction(msg, null);
                log.info("发送事务ID 检查：{}",JSONObject.toJSONString(sendResult));
            }
            while (true){
                Thread.sleep(5000);
                log.info("----- {} ", sdf.format(new Date()));
            }
        } catch (Exception e) {
            producer.shutdown();
            e.printStackTrace();
        }
    }
}
